package main

import (
	"bytes"
	"context"
	"encoding/binary"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"math/rand"
	"net/http"
	"net/url"
	"os"
	"strings"
	"sync"
	"time"

	"bufio"
	"strconv"

	"github.com/gorilla/websocket"
	"gopkg.in/yaml.v2"
)

type Config struct {
	Equipment struct {
		ID          string `yaml:"id"`
		Type        string `yaml:"type"`
		SensorCount int    `yaml:"sensor_count"`
	} `yaml:"equipment"`
	Server struct {
		WebSocketURL string `yaml:"websocket_url"`
		HTTPLoginURL string `yaml:"http_login_url"`
	} `yaml:"server"`
	Auth struct {
		Username string `yaml:"username"`
		Password string `yaml:"password"`
	} `yaml:"auth"`
	Streaming struct {
		IntervalMS           int `yaml:"interval_ms"`
		PositionIncrement    int `yaml:"position_increment"`
		ReconnectIntervalMin int `yaml:"reconnect_interval_minutes"`
		HeartbeatIntervalSec int `yaml:"heartbeat_interval_seconds"`
		RetryIntervalSec     int `yaml:"retry_interval_seconds"`
		CheckIntervalSec     int `yaml:"check_interval_seconds"`
	} `yaml:"streaming"`
	Token struct {
		RefreshIntervalSec  int `yaml:"refresh_interval_seconds"`  // token刷新间隔
		RefreshThresholdSec int `yaml:"refresh_threshold_seconds"` // token刷新阈值
		MaxRefreshRetries   int `yaml:"max_refresh_retries"`       // 最大刷新重试次数
	} `yaml:"token"`
}

var config Config

// const (
// 	equipmentId   = "798096803282026497"
// 	equipmentType = "1"
// 	sensorCount   = 40
// 	websocketUrl  = "ws://192.168.1.103:8808/api/v1/websocket"
// 	httpLoginUrl  = "http://192.168.1.103:8808/api/v1/system/login"
// )

type LoginRequest struct {
	Username string `json:"username"`
	Password string `json:"password"`
}

type LoginResponse struct {
	Code int    `json:"code"`
	Msg  string `json:"msg"`
	Data struct {
		Token string `json:"token"`
	} `json:"data"`
}

type Message struct {
	Event     string      `json:"event"`
	Data      interface{} `json:"data"`
	Code      int         `json:"code,omitempty"`
	ErrorMsg  string      `json:"errorMsg,omitempty"`
	Timestamp int64       `json:"timestamp,omitempty"`
}

type RegisterResponse struct {
	EquipmentId string `json:"equipmentId"`
	Message     string `json:"message"`
	Status      string `json:"status"`
	Code        string `json:"code"`
}

type UnregisterResponse struct {
	EquipmentId string `json:"equipmentId"`
	Message     string `json:"message"`
	Status      string `json:"status"`
	Code        string `json:"code"`
}

var (
	wsConn          *websocket.Conn
	token           string
	stopStream      chan struct{} // 用于控制数据流停止
	isStreaming     bool          // 标记是否正在推送数据流
	positionData    [8]uint32     // 存储位置数据
	lastPosition    uint32        // 记录最后一个位置的值
	positionMutex   sync.Mutex    // 添加互斥锁保护位置数据
	reconnectTicker *time.Ticker  // 用于定时重新连接
	lastReconnect   time.Time     // 记录最后一次重连时间
	reconnectMutex  sync.Mutex    // 保护重连操作
	tokenMutex      sync.RWMutex
	tokenExpireTime int64
	lastRefreshTime int64
)

// 设置默认配置值
func setDefaultConfig() {
	// 设置Streaming默认值
	if config.Streaming.IntervalMS <= 0 {
		config.Streaming.IntervalMS = 1000 // 默认1秒
	}
	if config.Streaming.PositionIncrement <= 0 {
		config.Streaming.PositionIncrement = 1
	}
	if config.Streaming.ReconnectIntervalMin <= 0 {
		config.Streaming.ReconnectIntervalMin = 5 // 默认5分钟
	}
	if config.Streaming.HeartbeatIntervalSec <= 0 {
		config.Streaming.HeartbeatIntervalSec = 30 // 默认30秒
	}
	if config.Streaming.RetryIntervalSec <= 0 {
		config.Streaming.RetryIntervalSec = 5 // 默认5秒
	}
	if config.Streaming.CheckIntervalSec <= 0 {
		config.Streaming.CheckIntervalSec = 10 // 默认10秒
	}

	// 设置Token默认值
	if config.Token.RefreshIntervalSec <= 0 {
		config.Token.RefreshIntervalSec = 3600 // 默认1小时
	}
	if config.Token.RefreshThresholdSec <= 0 {
		config.Token.RefreshThresholdSec = 300 // 默认5分钟
	}
	if config.Token.MaxRefreshRetries <= 0 {
		config.Token.MaxRefreshRetries = 3 // 默认3次
	}
}

func validateConfig() error {
	// 验证设备配置
	if config.Equipment.ID == "" {
		return fmt.Errorf("设备ID不能为空")
	}
	if config.Equipment.Type == "" {
		return fmt.Errorf("设备类型不能为空")
	}
	if config.Equipment.SensorCount <= 0 {
		return fmt.Errorf("传感器数量必须大于0")
	}

	// 验证服务器配置
	if config.Server.WebSocketURL == "" {
		return fmt.Errorf("WebSocket URL不能为空")
	}
	if _, err := url.Parse(config.Server.WebSocketURL); err != nil {
		return fmt.Errorf("WebSocket URL格式无效: %v", err)
	}
	if config.Server.HTTPLoginURL == "" {
		return fmt.Errorf("HTTP登录URL不能为空")
	}
	if _, err := url.Parse(config.Server.HTTPLoginURL); err != nil {
		return fmt.Errorf("HTTP登录URL格式无效: %v", err)
	}

	// 验证认证配置
	if config.Auth.Username == "" {
		return fmt.Errorf("用户名不能为空")
	}
	if config.Auth.Password == "" {
		return fmt.Errorf("密码不能为空")
	}

	// 设置默认值
	setDefaultConfig()

	return nil
}

func loadConfig() error {
	// 读取配置文件
	data, err := os.ReadFile("config.yaml")
	if err != nil {
		return fmt.Errorf("读取配置文件失败: %v", err)
	}

	err = yaml.Unmarshal(data, &config)
	if err != nil {
		return fmt.Errorf("解析配置文件失败: %v", err)
	}

	// 验证配置
	if err := validateConfig(); err != nil {
		return fmt.Errorf("配置验证失败: %v", err)
	}

	return nil
}

func login() error {
	loginReq := LoginRequest{
		Username: config.Auth.Username,
		Password: config.Auth.Password,
	}

	jsonData, err := json.Marshal(loginReq)
	if err != nil {
		return err
	}
	loginUrl := config.Server.HTTPLoginURL
	log.Printf("正在发送登录请求到: %s", loginUrl)
	resp, err := http.Post(loginUrl, "application/json", bytes.NewBuffer(jsonData))
	if err != nil {
		return fmt.Errorf("HTTP请求失败: %v", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("登录失败，状态码: %d, 响应: %s", resp.StatusCode, string(body))
	}

	var loginResp LoginResponse
	if err := json.NewDecoder(resp.Body).Decode(&loginResp); err != nil {
		return fmt.Errorf("解析响应失败: %v", err)
	}

	if loginResp.Code != 0 {
		return fmt.Errorf("登录失败: %s", loginResp.Msg)
	}

	if loginResp.Data.Token == "" {
		return fmt.Errorf("获取到的token为空")
	}

	token = loginResp.Data.Token
	return nil
}

func connectWebSocket() error {
	if wsConn != nil {
		wsConn.Close()
		wsConn = nil
	}

	dialer := websocket.Dialer{
		HandshakeTimeout: 10 * time.Second,
		ReadBufferSize:   32768, // 32KB
		WriteBufferSize:  32768, // 32KB
	}

	header := http.Header{}
	currentToken := getToken()
	if currentToken == "" {
		return fmt.Errorf("token为空，请先获取token")
	}
	header.Add("Authorization", "Bearer "+currentToken)

	// 设置连接超时上下文
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	// 使用带超时的连接
	conn, _, err := dialer.DialContext(ctx, config.Server.WebSocketURL, header)
	if err != nil {
		return fmt.Errorf("WebSocket连接失败: %v", err)
	}

	// 设置连接参数 - 符合终端设备规范
	conn.SetReadLimit(512 * 1024)                           // 512KB - MaxMessageSize
	conn.SetReadDeadline(time.Now().Add(90 * time.Second))  // 读取超时90秒
	conn.SetWriteDeadline(time.Now().Add(60 * time.Second)) // 普通消息写入超时60秒

	// 设置ping处理器 - WebSocket原生ping/pong
	conn.SetPingHandler(func(appData string) error {
		log.Printf("收到WebSocket原生Ping帧")
		// 立即回复pong
		err := conn.WriteControl(websocket.PongMessage, []byte{}, time.Now().Add(5*time.Second))
		if err != nil {
			log.Printf("回复WebSocket原生Pong帧失败: %v", err)
			return err
		}
		// 更新读取超时
		conn.SetReadDeadline(time.Now().Add(90 * time.Second))
		log.Printf("已回复WebSocket原生Pong帧")
		return nil
	})

	// 设置pong处理器 - WebSocket原生ping/pong
	conn.SetPongHandler(func(string) error {
		log.Printf("收到WebSocket原生Pong帧")
		// 更新读取超时
		conn.SetReadDeadline(time.Now().Add(90 * time.Second))
		return nil
	})

	wsConn = conn
	return nil
}

func sendMessage(event string, data interface{}) error {
	msg := Message{
		Event: event,
		Data:  data,
	}
	log.Printf("发送消息: %+v", msg)
	// 设置写入超时 - 普通消息60秒
	wsConn.SetWriteDeadline(time.Now().Add(60 * time.Second))
	err := wsConn.WriteJSON(msg)
	if err != nil {
		return err
	}
	// 更新读取超时 - 90秒
	wsConn.SetReadDeadline(time.Now().Add(90 * time.Second))
	return nil
}

func heartbeat() {
	ticker := time.NewTicker(time.Duration(config.Streaming.HeartbeatIntervalSec) * time.Second)
	defer ticker.Stop()

	var retryCount int
	const maxRetries = 3

	// 记录最后一次成功ping的时间
	lastSuccessfulPing := time.Now()

	for {
		select {
		case <-ticker.C:
			// 检查连接状态
			if wsConn == nil {
				log.Println("检测到连接为空，尝试重新连接...")
				if err := reconnect(); err != nil {
					log.Printf("重新连接失败: %v", err)
				}
				continue
			}

			// 检查是否超过最大无响应时间 - 90秒心跳超时
			if time.Since(lastSuccessfulPing) > 90*time.Second {
				log.Println("超过90秒心跳超时，尝试重新连接...")
				if err := reconnect(); err != nil {
					log.Printf("重新连接失败: %v", err)
				}
				continue
			}

			// 发送ping - 使用3秒发送超时避免阻塞数据流
			wsConn.SetWriteDeadline(time.Now().Add(3 * time.Second))
			if err := wsConn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(3*time.Second)); err != nil {
				retryCount++
				log.Printf("Ping发送失败: %v，重试次数: %d/%d", err, retryCount, maxRetries)

				// 检查是否是连接断开错误
				if websocket.IsUnexpectedCloseError(err) ||
					strings.Contains(err.Error(), "connection was aborted") ||
					strings.Contains(err.Error(), "wsasend") {
					log.Println("检测到连接断开，尝试重新连接...")
					if err := reconnect(); err != nil {
						log.Printf("重新连接失败: %v", err)
					} else {
						log.Println("重新连接成功")
						retryCount = 0
						lastSuccessfulPing = time.Now()
					}
				} else if retryCount >= maxRetries {
					log.Printf("Ping重试次数已达上限(%d)，尝试重新连接", maxRetries)
					if err := reconnect(); err != nil {
						log.Printf("重新连接失败: %v", err)
					} else {
						log.Println("重新连接成功")
						retryCount = 0
						lastSuccessfulPing = time.Now()
					}
				}
				continue
			}

			// 更新读取超时 - 90秒
			wsConn.SetReadDeadline(time.Now().Add(90 * time.Second))
			log.Println("Ping已发送")
			retryCount = 0
			lastSuccessfulPing = time.Now()
		}
	}
}

// 重新连接函数
func reconnect() error {
	reconnectMutex.Lock()
	defer reconnectMutex.Unlock()

	// 检查是否在冷却期内
	if time.Since(lastReconnect) < 30*time.Second {
		return fmt.Errorf("重连冷却中，请稍后再试")
	}

	// 关闭旧连接
	if wsConn != nil {
		wsConn.Close()
		wsConn = nil
	}

	// 重新连接
	if err := connectWebSocket(); err != nil {
		return fmt.Errorf("WebSocket连接失败: %v", err)
	}

	// 更新最后重连时间
	lastReconnect = time.Now()

	// 重新发送注册消息
	data := map[string]string{
		"equipmentId":   config.Equipment.ID,
		"equipmentType": config.Equipment.Type,
	}
	if err := sendMessage("register", data); err != nil {
		return fmt.Errorf("重新注册失败: %v", err)
	}

	return nil
}

// 计算CRC16-MODBUS
func calculateCRC(data []byte) uint16 {
	var crc uint16 = 0xFFFF
	for _, b := range data {
		crc ^= uint16(b)
		for i := 0; i < 8; i++ {
			if (crc & 0x0001) != 0 {
				crc = (crc >> 1) ^ 0xA001
			} else {
				crc = crc >> 1
			}
		}
	}
	return crc
}

// 生成二进制数据包
func generateBinaryData(number uint16) []byte {
	// 计算总长度：固定头部(28) + 数据包(32 + sensorCount*4*8) + CRC(2)
	totalLength := 28 + 32 + config.Equipment.SensorCount*4*8 + 2
	data := make([]byte, totalLength)

	// 1. code: 0-1: 0xaa55
	binary.LittleEndian.PutUint16(data[0:2], 0xaa55)

	// 2. sessionId: 2-13: "123456789000"
	copy(data[2:14], []byte("123456789000"))

	// 3. datetime: 14-21: 时间戳
	timestamp := uint64(time.Now().Unix())
	binary.LittleEndian.PutUint64(data[14:22], timestamp)

	// 4. number: 22-23: 序列号
	binary.LittleEndian.PutUint16(data[22:24], number)

	// 5. length: 24-27: 数据包长度
	dataLength := 32 + config.Equipment.SensorCount*4*8 // 位置数据(32) + 传感器数据(sensorCount*4*8)
	binary.LittleEndian.PutUint32(data[24:28], uint32(dataLength))

	// 6. 数据包: 28-28+n-1
	// 6.1 前32字节：8个int，表示位置（作为波形图的横坐标）
	positionMutex.Lock()
	defer positionMutex.Unlock()

	// 生成8个连续的位置值
	positions := make([]uint32, 8)
	for i := 0; i < 8; i++ {
		// 每个位置依次增加，形成连续的横坐标
		lastPosition += uint32(config.Streaming.PositionIncrement)
		positions[i] = lastPosition
		binary.LittleEndian.PutUint32(data[28+i*4:32+i*4], lastPosition)
	}

	// 6.2 后1280字节：8次采样，每次采样sensorCount个传感器的数据
	topFiveSensors := make([][]uint32, 5)
	for i := range topFiveSensors {
		topFiveSensors[i] = make([]uint32, 8)
	}

	for sample := 0; sample < 8; sample++ {
		startPos := 60 + sample*config.Equipment.SensorCount*4
		for sensor := 0; sensor < config.Equipment.SensorCount; sensor++ {
			// 每个传感器数据占4字节
			value := uint32(rand.Intn(1000))
			binary.LittleEndian.PutUint32(data[startPos+sensor*4:startPos+sensor*4+4], value)

			// 如果是前5个传感器，记录其值
			if sensor < 5 {
				topFiveSensors[sensor][sample] = value
			}
		}
	}

	// 7. CRC: 最后2字节
	// 计算除CRC外的所有数据的CRC
	crc := calculateCRC(data[:totalLength-2])
	// 使用大端序写入CRC
	binary.BigEndian.PutUint16(data[totalLength-2:], crc)

	// 打印数据包信息
	log.Printf("数据包信息:")
	log.Printf("  Code: 0x%04X", 0xaa55)
	log.Printf("  SessionId: %s", string(data[2:14]))
	log.Printf("  Timestamp: %d (%s)", timestamp, time.Unix(int64(timestamp), 0).Format("2006-01-02 15:04:05"))
	log.Printf("  Number: %d", number)
	log.Printf("  DataLength: %d", dataLength)
	log.Printf("  传感器数量: %d", config.Equipment.SensorCount)
	log.Printf("  采样次数: %d", 8)
	log.Printf("  位置数据: %v", positions)
	log.Printf("  CRC: 0x%04X", crc)
	log.Printf("  总包大小: %d 字节", totalLength)

	// 打印前5个传感器的8个位置的值
	log.Printf("前5个传感器的8个位置的值:")
	for i := 0; i < 5; i++ {
		log.Printf("  传感器 %d: %v", i+1, topFiveSensors[i])
	}

	// 打印整个数据包的内容
	log.Printf("数据包完整内容:")
	log.Printf("  头部(0-27):")
	log.Printf("    Code(0-1): 0x%04X", binary.LittleEndian.Uint16(data[0:2]))
	log.Printf("    SessionId(2-13): %s", string(data[2:14]))
	log.Printf("    Timestamp(14-21): %d", binary.LittleEndian.Uint64(data[14:22]))
	log.Printf("    Number(22-23): %d", binary.LittleEndian.Uint16(data[22:24]))
	log.Printf("    Length(24-27): %d", binary.LittleEndian.Uint32(data[24:28]))

	log.Printf("  位置数据(28-59):")
	positionStr := "    位置: "
	for i := 0; i < 8; i++ {
		positionStr += fmt.Sprintf("%d ", binary.LittleEndian.Uint32(data[28+i*4:32+i*4]))
	}
	log.Printf(positionStr)

	log.Printf("  传感器数据(60-%d):", totalLength-3)
	for sample := 0; sample < 8; sample++ {
		startPos := 60 + sample*config.Equipment.SensorCount*4
		currentSample := make([]uint32, config.Equipment.SensorCount)
		for sensor := 0; sensor < config.Equipment.SensorCount; sensor++ {
			currentSample[sensor] = binary.LittleEndian.Uint32(data[startPos+sensor*4 : startPos+sensor*4+4])
		}
		log.Printf("    采样%d: %v", sample+1, currentSample)
	}

	log.Printf("  CRC(1340-1341): 0x%04X", binary.BigEndian.Uint16(data[totalLength-2:]))

	return data
}

// 启动消息接收
func startMessageReceiver() {
	var reconnectDelay time.Duration = time.Duration(config.Streaming.RetryIntervalSec) * time.Second
	maxReconnectDelay := 5 * time.Minute
	reconnectAttempts := 0
	maxReconnectAttempts := 10

	for {
		// 检查连接状态
		if wsConn == nil {
			log.Println("检测到连接为空，尝试重新连接...")
			if err := reconnect(); err != nil {
				log.Printf("重新连接失败: %v", err)
				reconnectAttempts++
				if reconnectAttempts >= maxReconnectAttempts {
					log.Printf("达到最大重连次数(%d)，退出消息接收", maxReconnectAttempts)
					return
				}
				// 使用指数退避策略
				if reconnectDelay < maxReconnectDelay {
					reconnectDelay *= 2
				}
				log.Printf("等待 %v 后重试...", reconnectDelay)
				time.Sleep(reconnectDelay)
				continue
			}
			log.Println("重新连接成功")
			reconnectAttempts = 0
			reconnectDelay = time.Duration(config.Streaming.RetryIntervalSec) * time.Second

			// 重新发送注册消息
			data := map[string]string{
				"equipmentId":   config.Equipment.ID,
				"equipmentType": config.Equipment.Type,
			}
			if err := sendMessage("register", data); err != nil {
				log.Printf("重新注册失败: %v", err)
				time.Sleep(reconnectDelay)
				continue
			}
			log.Println("重新注册成功")
		}

		// 设置读取超时 - 90秒
		wsConn.SetReadDeadline(time.Now().Add(90 * time.Second))

		messageType, message, err := wsConn.ReadMessage()
		if err != nil {
			log.Printf("读取消息失败: %v", err)

			// 检查是否是连接关闭错误或超时错误
			if websocket.IsUnexpectedCloseError(err) ||
				strings.Contains(err.Error(), "connection was forcibly closed") ||
				strings.Contains(err.Error(), "wsarecv") ||
				strings.Contains(err.Error(), "connection was aborted") ||
				strings.Contains(err.Error(), "i/o timeout") {

				log.Println("检测到连接断开或超时，尝试重新连接...")
				// 关闭旧连接
				if wsConn != nil {
					wsConn.Close()
					wsConn = nil
				}

				// 尝试重新连接
				if err := reconnect(); err != nil {
					log.Printf("重新连接失败: %v", err)
					reconnectAttempts++
					if reconnectAttempts >= maxReconnectAttempts {
						log.Printf("达到最大重连次数(%d)，退出消息接收", maxReconnectAttempts)
						return
					}
					// 使用指数退避策略
					if reconnectDelay < maxReconnectDelay {
						reconnectDelay *= 2
					}
					log.Printf("等待 %v 后重试...", reconnectDelay)
					time.Sleep(reconnectDelay)
					continue
				}

				log.Println("重新连接成功")
				reconnectAttempts = 0
				reconnectDelay = time.Duration(config.Streaming.RetryIntervalSec) * time.Second

				// 重新发送注册消息
				data := map[string]string{
					"equipmentId":   config.Equipment.ID,
					"equipmentType": config.Equipment.Type,
				}
				if err := sendMessage("register", data); err != nil {
					log.Printf("重新注册失败: %v", err)
					time.Sleep(reconnectDelay)
					continue
				}
				log.Println("重新注册成功")
				continue
			}
			continue
		}

		// 重置重连计数和延迟
		reconnectAttempts = 0
		reconnectDelay = time.Duration(config.Streaming.RetryIntervalSec) * time.Second

		switch messageType {
		case websocket.TextMessage:
			var msg Message
			if err := json.Unmarshal(message, &msg); err != nil {
				log.Printf("解析JSON消息失败: %v", err)
				continue
			}
			log.Printf("收到消息: %+v", msg)

			// 处理不同类型的消息
			switch msg.Event {
			case "start":
				// 如果收到start消息且当前没有在推送数据流，则开始推送
				if !isStreaming {
					isStreaming = true
					go startStreaming()
					log.Println("收到服务器start指令，开始推送二进制数据流")
				} else {
					log.Println("已经在推送数据流，忽略start指令")
				}

			case "stop":
				// 如果收到stop消息且当前正在推送数据流，则停止推送
				if isStreaming {
					if stopStream != nil {
						close(stopStream)
						stopStream = nil
					}
					isStreaming = false
					log.Println("收到服务器stop指令，停止推送二进制数据流")
				} else {
					log.Println("当前没有在推送数据流，忽略stop指令")
				}

			case "register":
				// 处理注册响应
				// 服务器返回格式：Code在顶层，Data包含设备信息
				if msg.Code == 0 {
					log.Printf("设备注册成功: %+v", msg.Data)
				} else {
					log.Printf("设备注册失败: Code=%d, ErrorMsg=%s, Data=%+v", msg.Code, msg.ErrorMsg, msg.Data)
				}

			case "unregister":
				// 处理注销响应
				// 服务器返回格式：Code在顶层，Data包含设备信息
				if msg.Code == 0 {
					log.Printf("设备注销成功: %+v", msg.Data)
				} else {
					log.Printf("设备注销失败: Code=%d, ErrorMsg=%s, Data=%+v", msg.Code, msg.ErrorMsg, msg.Data)
				}

			default:
				log.Printf("收到未知事件: %s", msg.Event)
			}
		case websocket.BinaryMessage:
			log.Printf("收到二进制消息，大小: %d bytes", len(message))
		}
	}
}

// 开始发送数据流
func startStreaming() {
	stopStream = make(chan struct{})
	ticker := time.NewTicker(time.Duration(config.Streaming.IntervalMS) * time.Millisecond)
	defer ticker.Stop()

	var number uint16 = 0
	const (
		maxRetries     = 5                      // 最大重试次数
		initialBackoff = 100 * time.Millisecond // 初始退避时间
		maxBackoff     = 5 * time.Second        // 最大退避时间
	)

	for {
		select {
		case <-ticker.C:
			// 检查连接状态
			if wsConn == nil {
				log.Println("检测到连接为空，尝试重新连接...")
				if err := reconnect(); err != nil {
					log.Printf("重新连接失败: %v", err)
					isStreaming = false
					return
				}
				log.Println("重新连接成功")
				// 重新发送注册消息
				data := map[string]string{
					"equipmentId":   config.Equipment.ID,
					"equipmentType": config.Equipment.Type,
				}
				if err := sendMessage("register", data); err != nil {
					log.Printf("重新注册失败: %v", err)
					isStreaming = false
					return
				}
				log.Println("重新注册成功")
				continue
			}

			// 生成二进制数据
			binaryData := generateBinaryData(number)
			number++

			// 使用指数退避重试机制发送二进制数据
			backoff := initialBackoff
			var lastErr error
			for retry := 0; retry < maxRetries; retry++ {
				// 设置二进制数据写入超时 - 300秒
				wsConn.SetWriteDeadline(time.Now().Add(300 * time.Second))

				if err := wsConn.WriteMessage(websocket.BinaryMessage, binaryData); err != nil {
					lastErr = err
					log.Printf("发送二进制数据流失败(尝试 %d/%d): %v", retry+1, maxRetries, err)

					// 检查是否是连接断开错误
					if websocket.IsUnexpectedCloseError(err) ||
						strings.Contains(err.Error(), "connection was aborted") ||
						strings.Contains(err.Error(), "wsasend") ||
						strings.Contains(err.Error(), "i/o timeout") ||
						strings.Contains(err.Error(), "broken pipe") {

						log.Println("检测到连接问题，尝试重新连接...")
						if err := reconnect(); err != nil {
							log.Printf("重新连接失败: %v", err)
							isStreaming = false
							return
						}
						log.Println("重新连接成功")

						// 重新发送注册消息
						data := map[string]string{
							"equipmentId":   config.Equipment.ID,
							"equipmentType": config.Equipment.Type,
						}
						if err := sendMessage("register", data); err != nil {
							log.Printf("重新注册失败: %v", err)
							isStreaming = false
							return
						}
						log.Println("重新注册成功")

						// 重置退避时间
						backoff = initialBackoff
						continue
					}

					// 使用指数退避
					if retry < maxRetries-1 {
						log.Printf("等待 %v 后重试...", backoff)
						time.Sleep(backoff)
						backoff = time.Duration(float64(backoff) * 1.5)
						if backoff > maxBackoff {
							backoff = maxBackoff
						}
					}
					continue
				}

				// 发送成功
				log.Printf("二进制数据流已发送，数据大小: %d bytes, 序列号: %d", len(binaryData), number-1)
				lastErr = nil
				break
			}

			// 如果所有重试都失败
			if lastErr != nil {
				log.Printf("发送二进制数据流失败，已达到最大重试次数: %v", lastErr)
				isStreaming = false
				return
			}

		case <-stopStream:
			log.Println("停止发送数据流")
			isStreaming = false
			return
		}
	}
}

// 启动定时重新连接
func startReconnectTimer() {
	reconnectTicker = time.NewTicker(time.Duration(config.Streaming.ReconnectIntervalMin) * time.Minute)
	go func() {
		for {
			select {
			case <-reconnectTicker.C:
				// 检查是否在冷却期内
				reconnectMutex.Lock()
				if time.Since(lastReconnect) < 5*time.Minute {
					log.Println("重连冷却中，跳过本次重连")
					reconnectMutex.Unlock()
					continue
				}
				reconnectMutex.Unlock()

				log.Println("执行定时重新连接...")
				if err := reconnect(); err != nil {
					log.Printf("定时重新连接失败: %v", err)
				} else {
					log.Println("定时重新连接成功")
					// 更新最后重连时间
					reconnectMutex.Lock()
					lastReconnect = time.Now()
					reconnectMutex.Unlock()

					// 重新发送注册消息
					data := map[string]string{
						"equipmentId":   config.Equipment.ID,
						"equipmentType": config.Equipment.Type,
					}
					if err := sendMessage("register", data); err != nil {
						log.Printf("重新注册失败: %v", err)
					} else {
						log.Printf("重新注册成功: %+v", data)
					}
				}
			}
		}
	}()
}

// 更新token
func updateToken(newToken string, expireTime int64) {
	tokenMutex.Lock()
	defer tokenMutex.Unlock()
	token = newToken
	tokenExpireTime = expireTime
	lastRefreshTime = time.Now().Unix()
}

// 获取token
func getToken() string {
	tokenMutex.RLock()
	defer tokenMutex.RUnlock()
	return token
}

// 检查token是否需要刷新
func shouldRefreshToken() bool {
	tokenMutex.RLock()
	defer tokenMutex.RUnlock()

	// 如果token即将过期（在阈值时间内），则需要刷新
	return time.Now().Unix() > (tokenExpireTime - int64(config.Token.RefreshThresholdSec))
}

// 刷新token
func refreshToken() error {
	log.Printf("开始刷新token")

	// 准备登录数据
	loginData := LoginRequest{
		Username: config.Auth.Username,
		Password: config.Auth.Password,
	}

	jsonData, err := json.Marshal(loginData)
	if err != nil {
		return fmt.Errorf("序列化登录数据失败: %v", err)
	}

	// 发送登录请求
	resp, err := http.Post(config.Server.HTTPLoginURL, "application/json", bytes.NewBuffer(jsonData))
	if err != nil {
		return fmt.Errorf("登录请求失败: %v", err)
	}
	defer resp.Body.Close()

	// 检查响应状态码
	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("登录失败，状态码: %d, 响应: %s", resp.StatusCode, string(body))
	}

	// 解析响应
	var loginResp LoginResponse
	if err := json.NewDecoder(resp.Body).Decode(&loginResp); err != nil {
		return fmt.Errorf("解析响应失败: %v", err)
	}

	// 检查响应码
	if loginResp.Code != 0 {
		return fmt.Errorf("登录失败: %s", loginResp.Msg)
	}

	if loginResp.Data.Token == "" {
		return fmt.Errorf("获取到的token为空")
	}

	// 更新token
	updateToken(loginResp.Data.Token, time.Now().Unix()+2592000) // 30天过期
	log.Printf("token刷新成功")
	return nil
}

// 启动token监控
func startTokenMonitor() {
	ticker := time.NewTicker(time.Duration(config.Token.RefreshIntervalSec) * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			if shouldRefreshToken() {
				if err := refreshToken(); err != nil {
					log.Printf("token刷新失败: %v", err)
				}
			}
		}
	}
}

func main() {
	// 设置日志格式
	log.SetFlags(log.Ldate | log.Ltime | log.Lmicroseconds | log.Lshortfile)

	// 加载配置
	if err := loadConfig(); err != nil {
		log.Fatalf("加载配置失败: %v", err)
	}
	log.Println("配置加载成功")

	// 验证配置
	if err := validateConfig(); err != nil {
		log.Fatalf("配置验证失败: %v", err)
	}
	log.Println("配置验证成功")

	// 设置随机数种子
	rand.Seed(time.Now().UnixNano())

	// 首次登录获取token
	if err := refreshToken(); err != nil {
		log.Fatalf("初始登录失败: %v", err)
	}
	log.Println("初始登录成功")

	// 启动token监控
	go startTokenMonitor()

	// 连接WebSocket
	if err := connectWebSocket(); err != nil {
		log.Fatalf("WebSocket连接失败: %v", err)
	}
	log.Println("WebSocket连接成功")

	// 自动发送注册消息
	data := map[string]string{
		"equipmentId":   config.Equipment.ID,
		"equipmentType": config.Equipment.Type,
	}
	if err := sendMessage("register", data); err != nil {
		log.Printf("发送注册消息失败: %v", err)
	} else {
		log.Printf("注册消息已发送: %+v", data)
	}

	// 启动心跳
	go heartbeat()

	// 启动消息接收
	go startMessageReceiver()

	// 启动定时重新连接
	startReconnectTimer()

	// 检查是否在Docker容器中运行
	isDocker := os.Getenv("DOCKER_CONTAINER") == "true"

	if isDocker {
		// Docker容器模式：自动开始发送数据流
		log.Println("检测到Docker容器环境，自动开始发送数据流")
		isStreaming = true
		go startStreaming()

		// 保持程序运行
		select {}
	} else {
		// 本地模式：显示交互菜单
		reader := bufio.NewReader(os.Stdin)
		for {
			fmt.Println("\n请选择操作：")
			fmt.Println("1. 发送注册消息")
			fmt.Println("2. 发送取消消息")
			fmt.Println("3. 开始发送二进制数据流")
			fmt.Println("4. 停止发送数据流")
			fmt.Println("5. 退出")

			input, err := reader.ReadString('\n')
			if err != nil {
				log.Printf("读取输入失败: %v", err)
				continue
			}

			input = strings.TrimSpace(input)
			choice, err := strconv.Atoi(input)
			if err != nil {
				fmt.Println("无效的选择")
				continue
			}

			switch choice {
			case 1:
				data := map[string]string{
					"equipmentId":   config.Equipment.ID,
					"equipmentType": config.Equipment.Type,
				}
				if err := sendMessage("register", data); err != nil {
					log.Printf("发送注册消息失败: %v", err)
				} else {
					log.Printf("注册消息已发送: %+v", data)
				}
			case 2:
				data := map[string]string{
					"equipmentId":   config.Equipment.ID,
					"equipmentType": config.Equipment.Type,
				}
				if err := sendMessage("unregister", data); err != nil {
					log.Printf("发送取消消息失败: %v", err)
				} else {
					log.Printf("取消消息已发送: %+v", data)
				}
			case 3:
				if !isStreaming {
					isStreaming = true
					go startStreaming()
					log.Println("开始发送二进制数据流")
				} else {
					log.Println("已经在发送数据流")
				}
			case 4:
				if stopStream != nil {
					close(stopStream)
					stopStream = nil
					isStreaming = false
				}
			case 5:
				if stopStream != nil {
					close(stopStream)
				}
				if reconnectTicker != nil {
					reconnectTicker.Stop()
				}
				return
			default:
				fmt.Println("无效的选择")
			}
		}
	}
}
